package com.lg.distributed.transaction.message.sender;

import com.alibaba.fastjson.JSONObject;
import com.lg.distributed.transaction.message.constant.MQConstants;
import com.lg.distributed.transaction.message.coordinator.CompleteCorrelationData;
import com.lg.distributed.transaction.message.model.RabbitMetaMessage;
import com.lg.distributed.transaction.message.util.IdWorker;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * 消息的发送者
 *
 * @author by Mr. Li 2020/12/11 23:46
 */
@Component
@Slf4j
public class RabbitSender {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 设置对应的 db ，方便后调会调用应用
     *
     * @param correlation 持久化方案名称
     */
    public void setCorrelationData(String correlation) {
        rabbitTemplate.setCorrelationDataPostProcessor((message, correlationData) -> new CompleteCorrelationData(correlationData != null ? correlationData.getId() :
                null, correlation));
    }

    /**
     * 发送MQ信息
     *
     * @param rabbitMetaMessage Rabbit 元信息对象
     * @return 消息ID
     */
    public String send2DeadQueue(RabbitMetaMessage rabbitMetaMessage) {
        String msgId = rabbitMetaMessage.getMessageEntity().getMessageId();

        String result = JSONObject.toJSONString(rabbitMetaMessage.getPayload());

        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setContentType("application/json");
        Message message = new Message(result.getBytes(), messageProperties);
        message.getMessageProperties().setMessageId(msgId);

        try {
            // 消息发送后，会回调 MessagePostProcessor
            rabbitTemplate.convertAndSend(MQConstants.DLX_EXCHANGE,
                    MQConstants.DLX_ROUTING_KEY, message, new CorrelationData(msgId));
            log.info("消息进入到死信队列，一分钟之后重试，消息ID:{}", msgId);
            return msgId;
        } catch (AmqpException e) {
            throw new RuntimeException("发送RabbitMQ消息失败！", e);
        }
    }

    /**
     * 发送MQ信息
     *
     * @param rabbitMetaMessage Rabbit 元信息对象
     * @return 消息ID
     */
    public String send(RabbitMetaMessage rabbitMetaMessage) {
        String msgId = rabbitMetaMessage.getMessageEntity().getMessageId();

        MessagePostProcessor messagePostProcessor = message -> {
            // 设置消息持久化
            message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
            return message;
        };
        String result = JSONObject.toJSONString(rabbitMetaMessage.getPayload());

        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setContentType("application/json");
        Message message = new Message(result.getBytes(), messageProperties);
        message.getMessageProperties().setMessageId(msgId);

        try {
            // 消息发送后，会回调 MessagePostProcessor
            rabbitTemplate.convertAndSend(rabbitMetaMessage.getMessageEntity().getToExchange(),
                    rabbitMetaMessage.getMessageEntity().getRoutingKey(), message, messagePostProcessor, new CorrelationData(msgId));
            log.info("发送消息，消息ID:{}", msgId);
            return msgId;
        } catch (AmqpException e) {
            throw new RuntimeException("发送RabbitMQ消息失败！", e);
        }
    }

}
